/* Definitions for futexes (generic version).

   This file is part of khipu.

   khipu is free software: you can redistribute it and/or modify
   it under the terms of the GNU Lesser General Public License as published by
   the Free Software Foundation; either version 3 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU Lesser General Public License for more details.

   You should have received a copy of the GNU Lesser General Public License
   along with this program.  If not, see <https://www.gnu.org/licenses/>.  */

#include <mutex>
#include <condition_variable>
#include <chrono>
#include "../utils/lazy.hpp"

KP_DECLS_BEGIN

struct sleepq_bucket
{
  std::mutex lock;
  char pad[64];
  dlist queues;

  sleepq_bucket ()
    {
      this->queues.init_head ();
    }
};

struct sleepq
{
  std::condition_variable cv;
  uintptr_t key = 0;
  dlist link;
  dlist free_list;
  uint32_t nwaiters = 0;

  sleepq ()
    {
      this->link.init_unlinked ();
      this->free_list.init_head ();
    }
};

static inline sleepq*
alloc_sleepq ()
{
  sleepq *sq = (sleepq *)xmalloc (sizeof (*sq));
  try
    {
      new (sq) sleepq ();
    }
  catch (...)
    {
      xfree (sq);
      sq = nullptr;
    }

  return (sq);
}

static inline void
free_sleepq (sleepq *sq)
{
  xfree (sq);
}

static inline void
lend_sleepq (thread *tp, dlist *out)
{
  out->add (&tp->sleep_q->link);
  tp->sleep_q = nullptr;
}

static inline void
acquire_sleepq (thread *tp, dlist *node)
{
  auto it = node->prev->iterator (&sleepq::link);
  it->link.del ();
  tp->sleep_q = &*it;
}

static const int NBUCKETS = 128;
static lazy<sleepq_bucket> sleepq_buckets[NBUCKETS];

static_assert (alignof (atomic_t) % 4 == 0, "invalid alignment for atomic_t");

static int
futex_wait_impl (interpreter *interp, atomic_t *addr, int value, double *tp)
{
  uintptr_t key = (uintptr_t)addr >> 2;
  auto bucket = sleepq_buckets[key % NBUCKETS].ptr ();
  thread *self = as_thread (interp->thread);
  sleepq *sq = nullptr;
  int rv = THR_INTR;

  std::unique_lock<std::mutex> lk { bucket->lock };
  if (*addr != value)
    return (rv);

  for (auto it = bucket->queues.iterator (&sleepq::link); it.valid (); ++it)
    if (it->key == key)
      {
        sq = &*it;
        break;
      }

  if (!sq)
    {
      sq = self->sleep_q;
      sq->key = key;
      lend_sleepq (self, &bucket->queues);
    }
  else
    lend_sleepq (self, &sq->free_list);

  ++sq->nwaiters;
  if (!tp)
    sq->cv.wait (lk);
  else
    {
      std::chrono::nanoseconds ns ((int64_t)(*tp * 1e9));
      std::chrono::time_point<std::chrono::system_clock> tim (ns);
      if (sq->cv.wait_until (lk, tim) == std::cv_status::timeout)
        rv = THR_TIMEOUT;
    }

  acquire_sleepq (self, --sq->nwaiters ? sq->free_list.next : &sq->link);
  return (rv);
}

static void
futex_wake (atomic_t *ptr, bool wake_all)
{
  uintptr_t key = (uintptr_t)ptr >> 2;
  auto bucket = sleepq_buckets[key % NBUCKETS].ptr ();

  std::unique_lock<std::mutex> lk { bucket->lock };
  for (auto it = bucket->queues.iterator (&sleepq::link); it.valid (); ++it)
    if (it->key == key)
      {
        if (wake_all)
          it->cv.notify_all ();
        else
          it->cv.notify_one ();

        return;
      }
}

static bool
futex_init (interpreter *)
{
  for (int i = 0; i < NBUCKETS; ++i)
    {
      auto bucket = sleepq_buckets[i].ptr ();
      try
        {
          new (bucket) sleepq_bucket ();
        }
      catch (...)
        {
          return (false);
        }
    }

  return (true);
}

KP_DECLS_END
